package com.captjack.bigdata.hadoop.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * @author Capt Jack
 * @date 2017/12/13
 */
public class MapReduceTest {

    /**
     * 日志
     */
    public static final Logger logger = LoggerFactory.getLogger(MapReduceTest.class);

    /**
     * 四个泛型类型分别代表：
     * <p>
     * KeyIn        Mapper的输入数据的Key，这里是每行文字的起始位置（0,11,...）
     * <p>
     * ValueIn      Mapper的输入数据的Value，这里是每行文字
     * <p>
     * KeyOut       Mapper的输出数据的Key，这里是每行文字中的“年份”
     * <p>
     * ValueOut     Mapper的输出数据的Value，这里是每行文字中的“气温”
     */
    static class TempMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        /**
         * 键对应的值
         */
        private final IntWritable intWritable = new IntWritable(1);

        /**
         * map键
         */
        private Text text;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            text = new Text();
            super.setup(context);
        }

        /**
         * map方法是提供给map task进程来调用的，map task进程是每读取一行文本来调用一次我们自定义的map方法
         * map task在调用map方法时，传递的参数：
         * 一行的起始偏移量LongWritable作为key
         * 一行的文本内容Text作为value
         */
        @Override
        protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
            //拿到一行文本内容，转换成String 类型
            String line = value.toString();
            //将这行文本切分成单词
            String[] words = line.split(" ");
            //输出<单词，1>
            for (String word : words) {
                text.set(word);
                context.write(text, intWritable);
            }
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            text = null;
            super.cleanup(context);
        }

    }

    /**
     * 四个泛型类型分别代表：
     * <p>
     * KeyIn        Reducer的输入数据的Key，这里是每行文字中的“年份”
     * <p>
     * ValueIn      Reducer的输入数据的Value，这里是每行文字中的“气温”
     * <p>
     * KeyOut       Reducer的输出数据的Key，这里是不重复的“年份”
     * <p>
     * ValueOut     Reducer的输出数据的Value，这里是这一年中的“最高气温”
     */
    static class TempReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

        /**
         * 值类型
         */
        private IntWritable intWritable;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            intWritable = new IntWritable();
            super.setup(context);
        }

        /**
         * reduce方法提供给reduce task进程来调用
         * <p>
         * reduce task会将shuffle阶段分发过来的大量kv数据对进行聚合，聚合的机制是相同key的kv对聚合为一组
         * 然后reduce task对每一组聚合kv调用一次我们自定义的reduce方法
         * 比如：<hello,1><hello,1><hello,1><tom,1><tom,1><tom,1>
         * hello组会调用一次reduce方法进行处理，tom组也会调用一次reduce方法进行处理
         * 调用时传递的参数：
         * key：一组kv中的key
         * values：一组kv中所有value的迭代器
         */
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            //定义一个计数器
            int count = 0;
            //通过value这个迭代器，遍历这一组kv中所有的value，进行累加
            for (IntWritable value : values) {
                count += value.get();
            }
            intWritable.set(count);
            //输出这个单词的统计结果
            context.write(key, intWritable);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            intWritable = null;
            super.cleanup(context);
        }

    }

    /**
     * 传统map-reduce提交方式
     *
     * @param args 参数
     */
    public static void main(String[] args) {
        try {
            Configuration conf = new Configuration();
            conf.set("mapred.job.tracker", "local");
            conf.set("fs.defaultFS", "hdfs://CDH63:8020");
            // 获取作业对象
            Job wordCountJob = Job.getInstance(conf);
            // 重要：指定本job所在的jar包
            wordCountJob.setJarByClass(MapReduceTest.class);
            // 设置wordCountJob所用的mapper逻辑类为哪个类
            wordCountJob.setMapperClass(TempMapper.class);
            // 设置wordCountJob所用的reducer逻辑类为哪个类
            wordCountJob.setReducerClass(TempReducer.class);
            // 设置map阶段输出的kv数据类型
            wordCountJob.setMapOutputKeyClass(Text.class);
            wordCountJob.setMapOutputValueClass(IntWritable.class);
            // 设置最终输出的kv数据类型
            wordCountJob.setOutputKeyClass(Text.class);
            wordCountJob.setOutputValueClass(IntWritable.class);
            // 设置要处理的文本数据所存放的路径
            FileInputFormat.setInputPaths(wordCountJob, new Path("hdfs://CDH63:8020/tmp/cctest/"));
            FileOutputFormat.setOutputPath(wordCountJob, new Path("hdfs://CDH63:8020/tmp/ccout6/"));
            // 提交job给hadoop集群，等待执行完成
            wordCountJob.waitForCompletion(true);
            System.out.println("Finished");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}